package remoting.netty4.server;

import static java.util.concurrent.TimeUnit.MILLISECONDS;

import common.constants.CommonConstants;
import common.exception.RemotingException;
import common.model.Codec2;
import common.model.MagicUrl;
import common.util.ExecutorUtil;
import common.util.NetUtils;
import common.util.UrlUtils;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.DefaultThreadFactory;
import java.net.InetSocketAddress;
import java.util.Collection;
import java.util.HashSet;
import java.util.Map;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import remoting.AbstractServer;
import remoting.channel.Channel;
import remoting.channel.ChannelHandler;
import remoting.channel.ChannelHandlers;
import remoting.netty4.codec.NettyCodecAdapter;
import threadpools.ThreadPool;

public class NettyServer extends AbstractServer {

  private static final Logger logger = LoggerFactory.getLogger(NettyServer.class);
  private Map<String, Channel> channels;
  private ServerBootstrap bootstrap;

  private io.netty.channel.Channel channel;

  private EventLoopGroup bossGroup;
  private EventLoopGroup workerGroup;

  public NettyServer(final MagicUrl url, ChannelHandler handler, ThreadPool threadPool,
      Codec2 codec) throws RemotingException {
    super(url, ChannelHandlers
            .wrap(handler, ExecutorUtil.setThreadName(url, SERVER_THREAD_POOL_NAME), threadPool), codec);
  }

  @Override
  protected void doOpen() throws Throwable {
    bootstrap = new ServerBootstrap();

    bossGroup = new NioEventLoopGroup(1, new DefaultThreadFactory("NettyServerBoss", true));
    workerGroup = new NioEventLoopGroup(getUrl()
        .getPositiveParameter(CommonConstants.IO_THREADS_KEY, CommonConstants.DEFAULT_IO_THREADS),
        new DefaultThreadFactory("NettyServerWorker", true));

    final NettyServerHandler nettyServerHandler = new NettyServerHandler(getUrl(), this, getCodec());
    channels = nettyServerHandler.getChannels();

    bootstrap.group(bossGroup, workerGroup)
        .channel(NioServerSocketChannel.class)
        .childOption(ChannelOption.TCP_NODELAY, Boolean.TRUE)
        .childOption(ChannelOption.SO_REUSEADDR, Boolean.TRUE)
        .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
        .childHandler(new ChannelInitializer<NioSocketChannel>() {
          @Override
          protected void initChannel(NioSocketChannel ch) {
            int idleTimeout = UrlUtils.getIdleTimeout(getUrl());
            NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyServer.this);
            ch.pipeline()//.addLast("logging",new LoggingHandler(LogLevel.INFO))//for debug
                .addLast("decoder", adapter.getDecoder())
                .addLast("encoder", adapter.getEncoder())
                .addLast("server-idle-handler",
                    new IdleStateHandler(0, 0, idleTimeout, MILLISECONDS))
                .addLast("handler", nettyServerHandler);
          }
        });
    // bind
    ChannelFuture channelFuture = bootstrap.bind(getBindAddress());
    channelFuture.syncUninterruptibly();
    channel = channelFuture.channel();
  }

  @Override
  protected void doClose() throws Throwable {
    try {
      if (channel != null) {
        // unbind.
        channel.close();
      }
    } catch (Throwable e) {
      logger.warn(e.getMessage(), e);
    }
    try {
      Collection<Channel> channels = getChannels();
      if (channels != null && channels.size() > 0) {
        for (Channel channel : channels) {
          try {
            channel.close();
          } catch (Throwable e) {
            logger.warn(e.getMessage(), e);
          }
        }
      }
    } catch (Throwable e) {
      logger.warn(e.getMessage(), e);
    }
    try {
      if (bootstrap != null) {
        bossGroup.shutdownGracefully();
        workerGroup.shutdownGracefully();
      }
    } catch (Throwable e) {
      logger.warn(e.getMessage(), e);
    }
    try {
      if (channels != null) {
        channels.clear();
      }
    } catch (Throwable e) {
      logger.warn(e.getMessage(), e);
    }
  }

  @Override
  public Collection<Channel> getChannels() {
    Collection<Channel> chs = new HashSet<Channel>();
    for (Channel channel : this.channels.values()) {
      if (channel.isConnected()) {
        chs.add(channel);
      } else {
        channels.remove(NetUtils.toAddressString(channel.getRemoteAddress()));
      }
    }
    return chs;
  }

  @Override
  public Channel getChannel(InetSocketAddress remoteAddress) {
    return channels.get(NetUtils.toAddressString(remoteAddress));
  }

  @Override
  public boolean isBound() {
    return channel.isActive();
  }


  @Override
  public boolean canHandleIdle() {
    return true;
  }
}
